#pragma once

#include <iostream>
#include <memory>
#include <vector>
#include <mutex>
#include <functional>
#include <algorithm>
#include <queue>
#include <future>
#include <atomic>

#include "executor.h"

namespace good_cptl
{

template <typename T>
class Observable;

// 观察者基类（支持优先级）
template <typename T>
class Observer : public std::enable_shared_from_this<Observer<T>>
{
public:
  explicit Observer(uint32_t priority = 0) : priority_(priority) {}
  virtual ~Observer()
  {
    if (auto observable = observable_.lock()) {
      observable->remove_observer(this->shared_from_this());
    }
  }
  virtual void on_update(const T& data) = 0;
  int get_priority() const { return priority_; }
private:
  friend class Observable<T>;
  std::weak_ptr<Observable<T>> observable_;
  uint32_t priority_;
};


// 被观察者（生产者）
template<typename T>
class Observable : public std::enable_shared_from_this<Observable<T>>
{
public:
  using ObserverPtr = std::shared_ptr<Observer<T>>;
  // 加入观察者
  void add_observer(ObserverPtr observer, int priority = 0)
  {
    std::lock_guard<std::mutex> lock(mutex_);
    if (std::find(observers_.begin(), observers_.end(), observer) == observers_.end()) {
      return;
    }
    observer->priority_ = priority;
    observer->observable_ = this->shared_from_this();
    observers_.emplace_back(std::move(observer));

    // 按照优先级排序
    std::sort(observers_.begin(), observers_.end(),
              [](const ObserverPtr& lhs, const ObserverPtr& rhs) {
                return lhs->priority_ > rhs->priority_;
              });
  }
  // 移除观察者
  void remove_observer(const ObserverPtr& observer)
  {
    std::lock_guard<std::mutex> lock(mutex_);
    auto it = std::find_if(observers_.begin(), observers_.end(),
                            [observer](const ObserverPtr& o) { return o == observer; });
    if (it != observers_.end()) {
      observers_.erase(it);
    }
  }

  // 同步通知
  void notify(const T& value)
  {
    if (notifying_.exchange(true)) {
      return;
    }

    std::vector<ObserverPtr> observers_copy;
    {
      std::lock_guard<std::mutex> lock(mutex_);
      observers_copy = observers_;
    }
    for (auto& observer : observers_copy) {
      if (observer) {
        observer->on_update(value);
      }
    }
    notifying_.store(false);
  }

  void async_notify(const T& data)
  {
    std::function<void()> func = [this, data]() { this->notify(data); };
    cpu_thread_pool_exec_.add(func);
  }
private:
  std::vector<ObserverPtr> observers_;
  mutable std::mutex mutex_;
  std::atomic<bool> notifying_{false}; // 防止重入的标记

  // 异步通知
  CPUThreadPoolExecutor cpu_thread_pool_exec_;

};
}